package com.example.dobs.demo.flink.io;
import org.apache.flink.api.common.serialization.SimpleStringEncoder;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSink;
import org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.DefaultRollingPolicy;

import java.util.concurrent.TimeUnit;

public class FlinkWriteFileWithPrefix {
    public static void main(String[] args) throws Exception {
        // 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 创建一个简单的数据流
        DataStream<String> input = env.fromElements("Hello", "World", "Flink");

        // 定义文件输出路径和前缀
        String outputPath = "file:///home/dob/develop/tmp/output";
        String filePrefix = "my_prefix_";

        // 创建 StreamingFileSink
        StreamingFileSink<String> sink = StreamingFileSink
                .forRowFormat(new Path(outputPath), new SimpleStringEncoder<String>("UTF-8"))
                .withRollingPolicy(
                        DefaultRollingPolicy.builder()
                                .withRolloverInterval(TimeUnit.MINUTES.toMillis(15))
                                .withInactivityInterval(TimeUnit.MINUTES.toMillis(5))
                                .withMaxPartSize(1024 * 1024 * 1024)
                                .build())
                .withBucketCheckInterval(1000)
                .withOutputFileConfig(org.apache.flink.streaming.api.functions.sink.filesystem.OutputFileConfig
                        .builder()
                        .withPartPrefix(filePrefix)
                        .build())
                .build();

        // 将数据写入文件
        input.addSink(sink);

        // 执行作业
        env.execute("Flink Write File with Prefix");
    }
}
